package me.tuyou.function;

import org.apache.flink.api.common.operators.Order;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.GroupReduceOperator;
import org.apache.flink.api.java.operators.ReduceOperator;
import org.apache.flink.api.java.operators.UnsortedGrouping;
import org.apache.flink.api.java.tuple.Tuple3;


public class FirstNTest {

    public static void main(String[] args) throws Exception {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        DataSource<Tuple3<Integer, String, Double>> input = env.fromElements(
                Tuple3.of(1, "a", 1.1),
                Tuple3.of(1, "a", 1.0),
                Tuple3.of(2, "b", 2.3),
                Tuple3.of(2, "b", 2.2),
                Tuple3.of(3, "c", 3.9),
                Tuple3.of(3, "c", 3.4)
        );

        //first返回全局或者分组的前n个
        GroupReduceOperator<Tuple3<Integer, String, Double>, Tuple3<Integer, String, Double>> out1 = input.first(2);

        GroupReduceOperator<Tuple3<Integer, String, Double>, Tuple3<Integer, String, Double>> out2 = input.groupBy(1).first(1);

        GroupReduceOperator<Tuple3<Integer, String, Double>, Tuple3<Integer, String, Double>> out3 = input.groupBy(1).sortGroup(2, Order.ASCENDING).first(1);

        out1.print();
        System.out.println("--------");
        out2.print();
        System.out.println("--------");
        out3.print();

    }
}
